package com.atguigu.gmall.realtime.bootstrap

import java.util

import com.alibaba.fastjson.JSONObject
import com.atguigu.gmall.realtime.util.{MySqlUtil, MykafkaSender, RedisUtil}
import redis.clients.jedis.Jedis

object BootstrapDim {


  // 1  读mysql
  // 2   参考canal的格式 转换数据
  // 3  写kafka
  def main(args: Array[String]): Unit = {
    //0 获得维度表的清单
    val jedis: Jedis = RedisUtil.getJedisClient
    val dimTablesKey="DIM_TABLES"
    val dimTablesSet: util.Set[String] = jedis.smembers(dimTablesKey)
    jedis.close()

    // 1  读mysql
    import  collection.JavaConverters._
    for (dimTableName <- dimTablesSet.asScala ) {
       val sql=s"select * from $dimTableName"
       val jsonObjList: util.List[JSONObject] = MySqlUtil.queryList(sql)
      // 2   参考canal的格式 转换数据
       for ( jsonObj <- jsonObjList.asScala ) {
         val kafkaJsonObj = new JSONObject()
          val dataList = new util.ArrayList[JSONObject]()
         dataList.add(jsonObj)
         kafkaJsonObj.put("data",dataList)
         kafkaJsonObj.put("type","INSERT")
         kafkaJsonObj.put("table", dimTableName )
         println(kafkaJsonObj)
         // 3  写kafka  // 由于 引导程序 不会有一个键值反复操作的情况 所以不用考虑乱序问题  不用考虑局部分区的分区键
         MykafkaSender.send("ODS_BASE_DB_C", kafkaJsonObj.toJSONString)
       }
    }
  //  MykafkaSender.flush()  // 杯中酒
    MykafkaSender.close()// 杯中酒
  }

}
